package com.elinshaw.hbase.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

public final class HbaseUtils {

    private HbaseUtils() {
    }

    /**
     * 在HBase中创建一张表
     */
    public static void createTable() throws Exception {
        Configuration conf = HBaseConfiguration.create();
        //指定ZooKeeper集群地址
        conf.set("hbase.zookeeper.quorum", "datanode1:2181,datanode1:2181,datanode1:2181");
        //创建连接对象Connection
        Connection conn = ConnectionFactory.createConnection(conf);
        //得到数据库管理员对象
        Admin admin = conn.getAdmin();
        //创建表描述，并指定表名
        TableName tableName = TableName.valueOf("t2");
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
        //构建列族对象
        ColumnFamilyDescriptor family = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("f1")).build();
        //设置列族
        tableDescriptorBuilder.setColumnFamily(family);
        //创建表
        admin.createTable(tableDescriptorBuilder.build());
        System.out.println("create table success!!");
    }

    /**
     * 向表t1中添加三条数据
     */
    public static void putData() throws Exception {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "datanode1:2181,datanode1:2181,datanode1:2181");
        //创建连接对象Connection
        Connection conn = ConnectionFactory.createConnection(conf);
        //Table负责与记录相关的操作,如增删改查等
        TableName tableName = TableName.valueOf("t2");
        Table table = conn.getTable(tableName);
        // 设置rowkey
        Put put = new Put(Bytes.toBytes("row1"));
        // 添加列数据，指定列族、列名与列值
        put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("name"),
                Bytes.toBytes("elinshaw"));
        put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("age"),
                Bytes.toBytes("27"));
        put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("address"),
                Bytes.toBytes("chengdu"));

        Put put2 = new Put(Bytes.toBytes("row2"));// 设置rowkey
        // 添加列数据，指定列族、列名与列值
        put2.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("name"),
                Bytes.toBytes("xiaoming23333"));
        put2.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("age"),
                Bytes.toBytes("30"));
        put2.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("address"),
                Bytes.toBytes("beijing2"));

        // 设置rowkey
        Put put3 = new Put(Bytes.toBytes("row3"));
        //添加列数据，指定列族、列名与列值
        put3.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("age"),
                Bytes.toBytes("31"));
        put3.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("address"),
                Bytes.toBytes("beijing3"));

        //执行添加数据
        table.put(put);
        table.put(put2);
        table.put(put3);
        //释放资源
        table.close();
        System.out.println("put data success!!");
    }

    /**
     * 获取数据
     */
    public static void getData() throws Exception {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "datanode1:2181,datanode1:2181,datanode1:2181");
        //获得数据库连接
        Connection conn = ConnectionFactory.createConnection(conf);
        //获取Table对象，指定查询表名，Table负责与记录相关的操作,如增删改查等
        Table table = conn.getTable(TableName.valueOf("t2"));
        //创建Get对象，根据rowkey查询,rowkey=row1
        Get get = new Get("row1".getBytes());
        //查询数据，取得结果集
        Result r = table.get(get);
        //循环输出每个单元格的数据
        for (Cell cell : r.rawCells()) {
            //取得当前单元格所属的列族名称
            String family = new String(CellUtil.cloneFamily(cell));
            //取得当前单元格所属的列名称
            String qualifier = new String(CellUtil.cloneQualifier(cell));
            //取得当前单元格的列值
            String value = new String(CellUtil.cloneValue(cell));
            //输出结果
            System.out.println("列：" + family + ":" + qualifier + "—————值:" + value);
        }
    }

    /**
     * 删除表t2中行键为row1的一整条数据
     */
    public static void deleteData() throws Exception {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "datanode1:2181,datanode1:2181,datanode1:2181");
        Connection conn = ConnectionFactory.createConnection(conf);
        //获取Table对象，指定表名，Table负责与记录相关的操作,如增删改查等
        TableName tableName = TableName.valueOf("t2");
        Table table = conn.getTable(tableName);
        //创建删除对象Delete，根据rowkey删除一整条
        Delete delete = new Delete(Bytes.toBytes("row1"));
        table.delete(delete);
        //释放资源
        table.close();
        System.out.println("delete data success!!");


    }


}
